package cn._51doit.kafka.clients

import java.util.{Properties, UUID}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

object ProducerDemo {

  def main(args: Array[String]): Unit = {

    // 1 配置参数
    val props = new Properties()
    // 连接kafka节点
    props.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    props.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //配置ACKS（数据同步的应答）
    //0:渣男
    //-1/all:暖男，要结合retries
    //1:leader分区写log成功即可
    props.setProperty("acks", "1")
    //重试次数
    props.setProperty("retries", "100")
    //实在压缩方式（是在客户端先压缩）
    props.setProperty("compression.type", "snappy")
    //设置客户端数据缓存大小(从默认的32M改成了100M)
    props.setProperty("buffer.memory", 100*1024*1024 + "")
    //设置请求超时时间
    props.setProperty("request.timeout.ms", "45000")

    val topic = "test"

    // 2 kafka的生产者
    val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](props)

    for (i <- 911 to 920) {
      // 3 封装数据的对象

      //第一种情况：没有指定Key和分区，默认的策略就是轮询，将数据均匀写入多个分区中(不是写一条就切换分区，而是将多条数据批量写入到一个分区，达到一定大小再切换)
      //val record = new ProducerRecord[String, String](topic,"value-" + i)

      //第二种情况：将数据发送到指定的分区编号
      //val record = new ProducerRecord[String, String](topic, 1 , null,"myvalue:"+i)
      //val record = new ProducerRecord[String, String](topic, 1 , "doit","myvalue:"+i)

      //val partitionNum = i % 3  // 指定数据均匀写入3个分区中
      //val record = new ProducerRecord[String, String](topic, partitionNum, null,"myvalue:"+i)


      //第三中情况：不指定分区编号，指定key，   分区编号 = key.hasacode % topic的分区数量
      //val record = new ProducerRecord[String, String](topic , "doit","myvalue:"+i)

      //根据key的hashcode值模除以topic分区的数量，返回一个分区编号
      val record = new ProducerRecord[String, String](topic , UUID.randomUUID().toString ,"myvalue:中国666"+i)

      // 4 发送消息
      producer.send(record)

    }

    println("message send success")


    // 释放资源
    producer.close()
  }
}
